/*
 * Copyright (C) 2008 DVOSoft, Inc.
 *
 * All rights reserved.
 */
package dvo.infact;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.informatica.powercenter.sdk.SDKException;
import com.informatica.powercenter.sdk.repository.ICTWidgetInstance;
import com.informatica.powercenter.sdk.repository.IField;
import com.informatica.powercenter.sdk.repository.IGroup;
import com.informatica.powercenter.sdk.server.ct.PCTPartitionDriver;
import com.informatica.powercenter.sdk.server.ct.PCTTransformationDriver;

/**
 * The TransformationDriver that is used to configure the runtime instance of the
 * Pivot transform.
 *
 * We assume the following:
 * All inputs and outputs are of type String
 * The input fields have names of the form &lt;OUTPUT_FIELD_NAME&gt;@&lt;PIVOTED_ROW_INDEX&gt;
 *
 * For example, say the pivot transform is given four input columns called,
 * FIRST_NAME@0, FIRST_NAME@1, LAST_NAME@0, and LAST_NAME@1, it must be given
 * two output columns called FIRST_NAME, and LAST_NAME.
 *
 * At runtime, for each input row, the values (FIRST_NAME@0, LAST_NAME@0) and (FIRST_NAME@1, LAST_NAME@1)
 * would produce two rows.
 *
 * @author Vinayak R. Borkar &lt;vinayakb@blacktitan.com&gt;
 */
class PivotTransformationDriverImpl extends PCTTransformationDriver {
    private ICTWidgetInstance widgetInstance;
    private int[][] targetMatrix;

    PivotTransformationDriverImpl(ICTWidgetInstance widgetInstance) {
        this.widgetInstance = widgetInstance;
    }

    @Override
    public PCTPartitionDriver createPartitionDriver(int i) throws SDKException {
        return new PivotPartitionDriverImpl(targetMatrix);
    }

    @Override
    public void deinit() throws SDKException {

    }

    @Override
    public void init() throws SDKException {
        List<String> inSchema = new ArrayList<String>();
        List<String> outSchema = new ArrayList<String>();
        Map<String, Integer> outIndexMap = new HashMap<String, Integer>();
        Map<String, IField> fields = new HashMap<String, IField>();
        List<IGroup> groups = widgetInstance.getTable().getGroupList();
        for (IGroup grp : groups) {
            List<IField> fList = grp.getFieldList();
            for (IField f : fList) {
                String name = f.getName();
                if (!fields.containsKey(name)) {
                    fields.put(name, f);
                    if (name.indexOf('@') < 0) {
                        outSchema.add(name);
                        outIndexMap.put(name, outSchema.size() - 1);
                    } else {
                        inSchema.add(name);
                    }
                }
            }
        }
        if (inSchema.size() % outSchema.size() != 0) {
            throw new SDKException(
                    "Inputs and Output cardinality do not match up: inSize % outSize != 0");
        }
        targetMatrix = new int[inSchema.size() / outSchema.size()][outSchema
                .size()];
        for (int i = 0; i < inSchema.size(); ++i) {
            String inName = inSchema.get(i);
            int idx = inName.indexOf('@');
            String varName = inName.substring(0, idx);
            String inIndexStr = inName.substring(idx + 1);
            int inIndex = -1;
            try {
                inIndex = Integer.parseInt(inIndexStr);
            } catch (NumberFormatException e) {
                throw new SDKException(inName + " not a valid input port name");
            }
            int outIndex = -1;
            try {
                outIndex = outIndexMap.get(varName);
            } catch (NullPointerException e) {
                throw new SDKException(inName
                        + " does not match any out var from " + outSchema);
            }
            targetMatrix[inIndex][outIndex] = i;
        }
    }
}
